package com.bfxy.disruptor.common.producer;

import java.util.concurrent.locks.LockSupport;

import com.bfxy.disruptor.common.InsufficientCapacityException;
import com.bfxy.disruptor.common.Util;
import com.bfxy.disruptor.common.balance.Sequence;
import com.bfxy.disruptor.common.waitStraty.WaitStrategy;

abstract class SingleProducerSequencerPad extends AbstractProductor {
    protected long p1, p2, p3, p4, p5, p6, p7;

    public SingleProducerSequencerPad(int bufferSize, WaitStrategy waitStrategy) {
        super(bufferSize, waitStrategy);
    }
}

abstract class SingleProducerSequencerFields extends SingleProducerSequencerPad {
    public SingleProducerSequencerFields(int bufferSize, WaitStrategy waitStrategy) {
        super(bufferSize, waitStrategy);
    }

    /**
     * Set to -1 as sequence starting point
     */
    protected long nextValue = Sequence.INITIAL_VALUE;//-1
    protected long cachedValue = Sequence.INITIAL_VALUE;//-1
}

//单生产者使用的Sequencer
public final class SingleProducer extends SingleProducerSequencerFields {
    protected long p1, p2, p3, p4, p5, p6, p7;

    public SingleProducer(int bufferSize, final WaitStrategy waitStrategy) {
        super(bufferSize, waitStrategy);//4   com.bfxy.disruptor.common.BlockingWaitStrategy@6d78f375
    }

    @Override
    public boolean hasAvailableCapacity(final int requiredCapacity) {
        long nextValue = this.nextValue;
     // wrapPoint是一个临界序号，必须比当前最小的未消费序号还小
        long wrapPoint = (nextValue + requiredCapacity) - bufferSize;
     // 当前的最小未消费序号
        long cachedGatingSequence = this.cachedValue;

        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {
            long minSequence = Util.getMinimumSequence(gatingSequences, nextValue);
            this.cachedValue = minSequence;

            if (wrapPoint > minSequence) {
                return false;
            }
        }

        return true;
    }

    //生产者将元素添加到队列；在具体的实现上是通过next和publish方法实现的
    @Override
    public long next() {
        return next(1);//下一个
    }

    //由于生产者为单线程，因此不用考虑并发
    @Override
    public long next(int n) {
        if (n < 1) {
            throw new IllegalArgumentException("n must be > 0");
        }

        long nextValue = this.nextValue;//nextValue当前已经放的位置，初始值是-1。

        //当前位置加n，索引位置加n表示加完之后索引的值。索引之后的值-Size等于-1表示刚刚装满，小于-1表示可以继续装，大于-1表示不能装了。
        long nextSequence = nextValue + n;
        //wrapPoint是要放的位置
        long wrapPoint = nextSequence - bufferSize;// bufferSize是ringbuffer的长度。wrapPoint表示有没有超过容量。

        //cachedValue是现在消费者最小的序号，缓存，不用每次都去遍历消费者Util.getMinimumSequence(gatingSequences, nextValue)来获取最小序号。
        long cachedGatingSequence = this.cachedValue;//-1，cachedValue是缓存的value用于缓存优化的。

        //放的位置大于最小消费者的位置   或者  最小消费者位置大于之前放的位置，就死循环等待阻塞。
        if (wrapPoint > cachedGatingSequence || cachedGatingSequence > nextValue) {//判断队列是否已满
            long minSequence;
            //自旋锁操作：不满足条件就阻塞，满足条件就往下走，
            //Distruptor就是利用条件判断阻塞的自旋锁避免了加锁。
            //Util.getMinimumSequence：找到消费者中最小的序号值。minSequence是最小的消费者序号。
            //while：的意思是  如果生产者序号大于消费者中最小的序号，就挂起进行自旋操作。
            while (wrapPoint > (minSequence = Util.getMinimumSequence(gatingSequences, nextValue))) {
                LockSupport.parkNanos(1L); // TODO（下个版本做操作）: Use waitStrategy to spin? 挂起1纳秒，就是空轮询死循环。作者在此添加了注释，或许后续版本会采用WaitStrategy来实现；
            }

            this.cachedValue = minSequence;//cachedValue是最小的消费者序号
        }
        //满足条件就获取了这个索引值，返回这个索引值。
        this.nextValue = nextSequence;

        return nextSequence;
    }

    @Override
    public long tryNext() throws InsufficientCapacityException {
        return tryNext(1);
    }

    @Override
    public long tryNext(int n) throws InsufficientCapacityException {
        if (n < 1) {
            throw new IllegalArgumentException("n must be > 0");
        }

        if (!hasAvailableCapacity(n)) {
            throw InsufficientCapacityException.INSTANCE;
        }

        long nextSequence = this.nextValue += n;

        return nextSequence;
    }

    @Override
    public long remainingCapacity() {
        long nextValue = this.nextValue;

        long consumed = Util.getMinimumSequence(gatingSequences, nextValue);
        long produced = nextValue;
        return getBufferSize() - (produced - consumed);
    }

    @Override
    public void claim(long sequence) {
        this.nextValue = sequence;
    }

    @Override
    public void publish(long sequence) {
        // 设置当前生产者的sequence，表示这个位置已经产生了元素，可以进行消费了，然后唤醒等待的消费者。
        cursor.set(sequence);//游标，初始值是-1.
        waitStrategy.signalAllWhenBlocking();//阻塞就唤醒，waitStrategy是new Disruptor时候传进来的。
    }

    @Override
    public void publish(long lo, long hi) {
        publish(hi);
    }

    @Override
    public boolean isAvailable(long sequence) {
        return sequence <= cursor.get();
    }

    @Override
    public long getHighestPublishedSequence(long lowerBound, long availableSequence) {
        return availableSequence;
    }
}
